#include "epoll.h"
#include "threadpool.hpp"
#include "requestData.h"
#include "util.h"
#include "timer.h"

#include <sys/epoll.h>
#include <queue>
#include <sys/time.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <stdio.h>
#include <string.h>
#include <cstdlib>
#include <iostream>
#include <vector>
#include <unistd.h>
#include <signal.h>
#include <assert.h>

using namespace std;

const int THREADPOOL_THREAD_NUM = 4;
const int QUEUE_SIZE = 65535;
const int LISTENQ = 1024;

const int PORT = 8888;

static int pipefd[2];
struct epoll_event *events;
bool stop_server = false;

extern priority_queue<mytimer*, vector<mytimer*>, timerCmp> timer_que;
extern pthread_mutex_t heap_locker;

void acceptConnection(int listen_fd, int epoll_fd);

int socket_bind_listen(int port) {
    if (port < 1024 || port > 65535)
        return -1;

    int listen_fd = 0;
    if ((listen_fd = socket(AF_INET, SOCK_STREAM, 0)) == -1)
        return -1;

    // 消除bind时"Address already in use"错误
    int optval = 1;
    if (setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval)) == -1)
        return -1;

    // 设置服务器IP和Port，和监听描述副绑定
    struct sockaddr_in server_addr;
    bzero((char *) &server_addr, sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    server_addr.sin_port = htons((unsigned short) port);
    if (bind(listen_fd, (struct sockaddr *) &server_addr, sizeof(server_addr)) == -1)
        return -1;

    if (listen(listen_fd, LISTENQ) == -1)
        return -1;

    if (listen_fd == -1) {
        close(listen_fd);
        return -1;
    }

    return listen_fd;
}

void acceptConnection(int listen_fd, int epoll_fd) {
    struct sockaddr_in client_addr;
    memset(&client_addr, 0, sizeof(struct sockaddr_in));
    socklen_t client_addr_len = 0;
    int accept_fd = 0;
    accept_fd = accept(listen_fd, (struct sockaddr *) &client_addr, &client_addr_len);
    if (accept_fd < 0) {
        printf("err!");
    }

    setNonBlocking(accept_fd);

    requestData *req_info = new requestData(epoll_fd, accept_fd);

    // 文件描述符可以读，边缘触发(Edge Triggered)模式，保证一个socket连接在任一时刻只被一个线程处理
    __uint32_t _epo_event = EPOLLIN | EPOLLET | EPOLLONESHOT | EPOLLRDHUP;
    epoll_add(epoll_fd, accept_fd, static_cast<void *>(req_info), _epo_event);
}

int epoll_init() {
    int epoll_fd = epoll_create(LISTENQ + 1);
    assert(epoll_fd != -1);
    events = new epoll_event[MAXEVENTS];
    return epoll_fd;
}

void pipefd_init(int epoll_fd) {
    int ret = socketpair(PF_UNIX, SOCK_STREAM, 0, pipefd); //初始化管道pipefd
    assert(ret != -1);
    setNonBlocking(pipefd[1]);

    requestData *req = new requestData();
    req->setFd(pipefd[0]);
    epoll_add(epoll_fd, pipefd[0], static_cast<void *>(req), EPOLLIN | EPOLLET);
}
/**
 * 函数逻辑：收到信号后就往管道写信号值
 * @param sig
 */
void sig_handler(int sig) {
    /*保留errno，在最后恢复，保证函数可重入性*/
    int save_errno = errno;
    int msg = sig;
    int ret = send(pipefd[1], (char *) &msg, 1, 0);
    errno = save_errno;
}

void add_sig(int sig) {
    struct sigaction sa;
    memset(&sa, '\0', sizeof(sa));
    sa.sa_handler = sig_handler;
    sa.sa_flags |= SA_RESTART;
    sigfillset(&sa.sa_mask);
    assert(sigaction(sig, &sa, NULL) != -1);
}

void handle_timer(){ //处理定时器超时事件
    pthread_mutex_lock(&heap_locker);
    while(!timer_que.empty()){ //定时器堆非空时
        mytimer* timer = timer_que.top();
        if(timer->is_deleted() || timer->is_expire()){
            timer_que.pop();
            delete timer;
            continue;
        }
        break; //此时定时器有效
    }
    pthread_mutex_unlock(&heap_locker);
}
int main() {
    int epoll_fd = epoll_init();
    pipefd_init(epoll_fd);
    pthread_mutex_init(&heap_locker, NULL); //初始化堆的互斥锁
    /*设置信号处理函数*/
    add_sig(SIGHUP);
    add_sig(SIGINT);
    add_sig(SIGPIPE);

    threadpool<requestData> *pool = NULL;
    try {
        pool = new threadpool<requestData>(THREADPOOL_THREAD_NUM, QUEUE_SIZE); //4和65535
    } catch (...) {
        perror("threadpool init failed"); //把一个描述性错误消息输出到标准错误 stderr
        return -1;
    }

    int listen_fd = socket_bind_listen(PORT);
    if (listen_fd < 0) {
        perror("socket bind failed");
        return 1;
    }
    setNonBlocking(listen_fd);

    requestData *req = new requestData();
    req->setFd(listen_fd);
    epoll_add(epoll_fd, listen_fd, static_cast<void *>(req), EPOLLIN | EPOLLET); //epoll以request data为一个处理对象

    while (!stop_server) {
        int events_num = epoll_wait(epoll_fd, events, MAXEVENTS, -1); //-1代表阻塞调用
        if (events_num < 0 && errno != EINTR) {
            perror("epoll failure");
            break;
        }

        for (int i = 0; i < events_num; i++) {
            requestData* request = static_cast<requestData *>(events[i].data.ptr);
            int fd = request->getFd();

            /*如果就绪的是listenfd，代表新连接到来*/
            if (fd == listen_fd) {
                acceptConnection(listen_fd, epoll_fd);
            }
            /*如果就绪的是pipefd[0]，则处理信号*/
            else if ((fd == pipefd[0]) && (events[i].events & EPOLLIN)) {
                char signals[1024];
                int ret = recv(pipefd[0], signals, sizeof(signals), 0);
                if (ret == -1) {
                    continue;
                } else if (ret == 0) {
                    continue;
                } else {
                    for (int i = 0; i < ret; ++i) {
                        switch (signals[i]) {
                            case SIGINT: {
                                stop_server = true;
                                printf("stop server!\n");
                                break;
                            }
                            case SIGHUP: {
                                continue;
                                break;
                            }
                            case SIGPIPE: {
                                //SIG_IGN(SIGPIPE); 
                                break;
                            }
                        }
                    }
                }
            }
            /*有异常直接关闭连接*/
            else if (events[i].events & (EPOLLERR | EPOLLHUP | EPOLLRDHUP)) {
                printf("error event\n");
                delete request; //有异常，直接关闭客户连接
                continue;
            }
            /*最后则是处理请求*/
            else if (events[i].events & EPOLLIN) {
                bool rc = pool->append(request);
                if(rc == false){
                    //任务队列满，释放连接
                    delete request;
                    printf("server que is full!\n");
                }
            }
        }
        handle_timer(); //处理定时事件
    }
    close(epoll_fd);
    close(listen_fd);
    delete pool;
    return 0;
}